package com.bindatax.core.task.operate.impl;

import com.bindatax.core.log.LogOutManager;
import com.bindatax.core.model.ValueMetaData;
import com.bindatax.core.sqlformatter.SqlFormatter;
import com.bindatax.core.task.ColumnMetaData;
import com.bindatax.core.task.DataTask;
import com.bindatax.core.database.operate.BaseDatabaseOperate;
import com.bindatax.core.database.operate.DatabaseOperateInterface;
import com.bindatax.core.dspool.DataTaskConnManager;
import com.bindatax.core.util.DatabaseUtil;
import org.pentaho.di.core.database.BaseDatabaseMeta;
import org.pentaho.di.core.row.ValueMetaInterface;

import java.sql.*;
import java.util.*;

public class OracleDatabaseOperate extends BaseDatabaseOperate implements DatabaseOperateInterface {

    private final static String SQL_TYPE_CLOB = "clob";
    private final static String SQL_TYPE_BLOB = "blob";

    @Override
    public String getDatabaseType() {
        return BaseDatabaseMeta.DATABASE_ORACLE_PLUGINID;
    }

    @Override
    public String getLimitSql(String tableName) {
        return "SELECT * FROM " + tableName + " WHERE ROWNUM < 2";
    }

    @Override
    public boolean checkTransValue(ValueMetaInterface commValueMeta) {
        String sqlTypeDesc = commValueMeta.getSqlTypeDesc();
        return SQL_TYPE_CLOB.equalsIgnoreCase(sqlTypeDesc) || SQL_TYPE_BLOB.equalsIgnoreCase(sqlTypeDesc);
    }

    @Override
    public Object transValue(ResultSet sourceResultSet,ValueMetaInterface commValueMeta) throws Exception{
        String resColName = commValueMeta.getName();
        String sqlTypeDesc = commValueMeta.getSqlTypeDesc();
        Object value = null;
        if (SQL_TYPE_CLOB.equalsIgnoreCase(sqlTypeDesc)) {
            Clob clob = sourceResultSet.getClob(resColName);
            value = DatabaseUtil.oracleClob2Str(clob);
        } else if (SQL_TYPE_BLOB.equalsIgnoreCase(sqlTypeDesc)) {
            Blob blob = sourceResultSet.getBlob(resColName);
            if (blob != null) {
                value = blob.getBytes(1, (int) blob.length());
            }
        }
        return value;
    }

    @Override
    public boolean checkDatabaseType(String databaseProductName, String databaseProductVersion) {
        if(databaseProductName == null){
            return false;
        }
        databaseProductName = databaseProductName.toLowerCase();
        return databaseProductName.contains(getDatabaseType().toLowerCase());
    }

    @Override
    public String[] getPrimaryKeys(DataTask thisDataTask) {
        String[] targetIds = null;
        // 如果没有设置ID，则尝试自动获取目标库表的ID字段
        LOG.info("没有设置主键列表但设置自动检测，现在对目标表获取主键信息。");
        String tableName = thisDataTask.getTargetFullTableName();
        String schemaName = tableName.substring(0, tableName.lastIndexOf("."));
        tableName = tableName.substring(tableName.lastIndexOf(".") + 1);
        String getPkSql = "SELECT\n" +
                "\tcol.column_name,\n" +
                "\tcon.constraint_type\n" +
                "FROM\n" +
                "\tdba_constraints con,\n" +
                "\tdba_cons_columns col \n" +
                "WHERE\n" +
                "\tcon.constraint_name = col.constraint_name \n" +
                "\tAND (con.constraint_type = 'P' OR con.constraint_type = 'U')\n" +
                "\tAND col.OWNER = '" + schemaName + "'\n" +
                "\tAND col.table_name = '" + tableName + "'";
        LOG.info("注意，没有设置入库目标表主键字段，准备检查目标表是否存在主键或唯一字段。"
                + LogOutManager.CR + "查询SQL："
                + LogOutManager.CR + getPkSql + LogOutManager.CR);

        Connection targetConn = null;
        PreparedStatement targetStatement = null;
        ResultSet targetResultSet = null;

        try {
            String targetConnKey = thisDataTask.getTargetConnKey();
            targetConn = DataTaskConnManager.getConnection(targetConnKey);
            targetStatement = targetConn.prepareStatement(getPkSql);
            targetResultSet = targetStatement.executeQuery();

            Map<String, List<String>> pkInfoMap = new HashMap<String, List<String>>();
            while (targetResultSet.next()) {
                String colName = targetResultSet.getString(1);
                String type = targetResultSet.getString(2);
                List<String> colList = pkInfoMap.computeIfAbsent(type, v -> new ArrayList<String>());
                colList.add(colName);
            }
            if (!pkInfoMap.isEmpty()) {
                //如果存在主键字段，则使用主键进行更新，否则尝试使用唯一列字段
                List<String> pkList = pkInfoMap.get("P");
                if (pkList != null && !pkList.isEmpty()) {
                    targetIds = new String[]{pkList.get(0)};
                } else {
                    List<String> uniqueList = new ArrayList<String>();
                    Collection<List<String>> values = pkInfoMap.values();
                    for (List<String> uList : values) {
                        uniqueList.addAll(uList);
                    }
                    if (!uniqueList.isEmpty()) {
                        targetIds = new String[uniqueList.size()];
                        for (int i = 0; i < targetIds.length; i++) {
                            targetIds[i] = uniqueList.get(i);
                        }
                    }
                }
                if(targetIds != null){
                    String msg = LogOutManager.CR + "注意，没有设置入库目标表[" + tableName + "]主键字段，" + LogOutManager.CR + "但检测到目标表存在主键或唯一字段，因此已自动设置更新/写入条件字段。目标表的主键字段为：" + LogOutManager.CR;
                    for (int i = 0; i < targetIds.length; i++) {
                        msg += "主键或唯一字段[" + (i + 1) + "]：" + targetIds[i] + LogOutManager.CR;
                    }
                    LOG.info(msg);
                }
            } else {
                LOG.info("表：" + tableName + " 不存在主键或唯一字段。");
            }
        } catch (Exception e) {
            LOG.error("获取表的主键或唯一字段列表信息失败！", e);
        } finally {
            DataTaskConnManager.close(targetResultSet, targetStatement, targetConn);
        }
        return targetIds;
    }

    @Override
    public String getWriteSql(DataTask thisDataTask) {
        String[] targetIds = thisDataTask.getTargetIds();
        ColumnMetaData columnMetaData = thisDataTask.getColumnMetaData();
        String sql;
        if (targetIds != null) {// 如果存在ID，则使用 更新或写入 方式
            StringBuilder mergeSql = new StringBuilder();
            StringBuilder updateSql = new StringBuilder();//更新SQL
            StringBuilder insertColNameSql = new StringBuilder();//写入SQL
            StringBuilder insertColZwfSql = new StringBuilder();//占位符SQL
            // 构建SQL
            mergeSql.append("MERGE INTO ").append(thisDataTask.getTargetFullTableName())
                    .append(" T1 USING ( SELECT ");
            for (int i = 0; i < targetIds.length; i++) {//构建ID列联合查询值
                mergeSql.append("? AS ").append(targetIds[i]).append(",");
            }
            mergeSql.setLength(mergeSql.length() - 1);
            mergeSql.append(" FROM DUAL ) T2 ON ( ");
            for (int i = 0; i < targetIds.length; i++) {//构建ID列联合查询值
                mergeSql.append("T1.").append(targetIds[i]).append("=T2.").append(targetIds[i]).append(" ").append("AND ");
            }
            mergeSql.setLength(mergeSql.length() - "AND ".length());
            mergeSql.append(") WHEN MATCHED THEN ");
            updateSql.append(" UPDATE SET ");

            Set<String> targetIdSet = new HashSet<>(targetIds.length);
            Collections.addAll(targetIdSet, targetIds);

            List<ValueMetaData> targetColumnNameList = columnMetaData.getTargetColumnNameList();
            int targetMetaSize = targetColumnNameList.size();
            for (int i = 0; i < targetMetaSize; i++) {
                ValueMetaData targetMeta = targetColumnNameList.get(i);
                String targetColName = targetMeta.getName();
                if (!targetIdSet.contains(targetColName)) {//update不能存在主键字段
                    updateSql.append(" T1.").append(targetColName).append("=?,");
                }
                insertColNameSql.append(targetColName).append(",");
                insertColZwfSql.append("?,");
            }

            updateSql.setLength(updateSql.length() - 1);
            insertColNameSql.setLength(insertColNameSql.length() - 1);
            insertColZwfSql.setLength(insertColZwfSql.length() - 1);
            mergeSql.append(updateSql)
                    .append(" WHEN NOT MATCHED THEN ")
                    .append(" INSERT ( ").append(insertColNameSql).append(" ) VALUES ( ")
                    .append(insertColZwfSql).append(" )");
            sql = mergeSql.toString();
            sql = SqlFormatter.format(sql);
            LOG.info("开始执行写入数据（已经赋值主键字段，使用更新或者写入方式）。"
                    + LogOutManager.CR + "编译SQL："
                    + LogOutManager.CR + sql + LogOutManager.CR);
        } else {// 否则直接写入
            // 拼装sql
            StringBuilder insertSql = new StringBuilder();//写入数据sql
            StringBuilder colNameSql = new StringBuilder();//列名SQL
            StringBuilder zwfSql = new StringBuilder();//占位符SQL

            List<ValueMetaData> targetColumnNameList = columnMetaData.getTargetColumnNameList();
            for (int i = 0; i < targetColumnNameList.size(); i++) {
                ValueMetaData targetMeta = targetColumnNameList.get(i);
                colNameSql.append(targetMeta.getName()).append(",");
                zwfSql.append("?,");
            }
            colNameSql.setLength(colNameSql.length() - 1);
            zwfSql.setLength(zwfSql.length() - 1);

            insertSql.append("INSERT INTO ")
                    .append(thisDataTask.getTargetFullTableName())
                    .append("(").append(colNameSql)
                    .append(") VALUES(").append(zwfSql).append(")");
            sql = insertSql.toString();
            sql = SqlFormatter.format(sql);
            LOG.info("开始执行写入数据（没有设置主键值，使用直接insert写入方式）。" + LogOutManager.CR + "编译SQL：" + LogOutManager.CR + sql);
        }
        return sql;
    }
}
